import com.zuikaku.pojo.Order;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.math.BigDecimal;
import java.util.UUID;

public class ReduceAPIDemo {
    public static void main(String[] args) {
        //创建环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironment();
        environment.setParallelism(1);
        //根据order获取source
        DataStreamSource<Order> orderDS = environment.fromElements(new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1003,new BigDecimal("6999.99")),
                new Order(UUID.randomUUID().toString(),"iPhone 15 Pro Max",1002,new BigDecimal("6979.99")),
                new Order(UUID.randomUUID().toString(),"HuaWei Mate60",1001,new BigDecimal("8999.99")),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1004,new BigDecimal("1999.99")),
                new Order(UUID.randomUUID().toString(),"Nokia X1",1005,new BigDecimal("1969.99")));
        //先按itemName进行分组
        KeyedStream<Order, String> keyedStream = orderDS.keyBy(new KeySelector<Order, String>() {
            @Override
            public String getKey(Order order) throws Exception {
                return order.getItemName();
            }
        });
        DataStream<Order> reduceDs = keyedStream.reduce(new ReduceFunction<Order>() {
            @Override
            public Order reduce(Order left, Order right) throws Exception {
                right.setPrice(left.getPrice().add(right.getPrice()));
                return right;
            }
        });
        reduceDs.print("订单按商品累加金额后");

        try {
            environment.execute("reduce job");
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
